use std::time::Duration;

use anyhow::Result;
use chrono::Local;
use sea_orm::DbConn;
use tokio::{
    sync::{broadcast, mpsc},
    time,
};
use tracing::{debug, error, info};

use crate::{
    apis::AriaRpc,
    models::{AriaParams, Category, Status, Task},
    repo::TaskRepo,
    state::AppState,
};

pub async fn start(state: AppState) {
    tokio::spawn(async move {
        let mut interval = time::interval(Duration::from_secs(10));

        loop {
            interval.tick().await;

            query_seeding_tasks(&state).await;

            let count = match query_active_tasks(&state).await {
                Ok(c) => c,
                Err(_) => continue,
            };

            if count < 5 {
                query_waiting_tasks(&state, 5 - count).await;
            }
        }
    });
}

async fn query_active_tasks(state: &AppState) -> Result<u64> {
    let list = match TaskRepo::query_active_tasks(&state.db).await {
        Ok(list) => list,
        Err(e) => {
            error!("query_active_tasks error: {}", e);
            return Err(e);
        }
    };

    if list.is_empty() {
        return Ok(0);
    }

    let count = list.len() as u64;

    for gid in list {
        if let Err(e) = state.submitter.send(gid).await {
            error!("submitter.send error: {}", e);
        }
    }

    Ok(count)
}

async fn query_seeding_tasks(state: &AppState) {
    let list = match TaskRepo::query_seeding_tasks(&state.db).await {
        Ok(list) => list,
        Err(e) => {
            error!("query_seeding_tasks error: {}", e);
            return;
        }
    };

    if list.is_empty() {
        return;
    }

    for gid in list {
        if let Err(e) = state.submitter.send(gid).await {
            error!("submitter.send error: {}", e);
        }
    }
}

pub type Submitter = mpsc::Sender<String>;
pub type Notifier = broadcast::Sender<Task>;

pub async fn start_queue(db: DbConn, aria_rpc: AriaRpc) -> (Submitter, Notifier) {
    let (tx, rx) = mpsc::channel::<String>(1024);
    let (tx2, _) = broadcast::channel::<Task>(1024);

    let notifier = tx2.clone();
    tokio::spawn(async move {
        deal(rx, db, aria_rpc, notifier).await;
    });

    (tx, tx2)
}

async fn deal(mut rx: mpsc::Receiver<String>, db: DbConn, aria_rpc: AriaRpc, notifier: Notifier) {
    while let Some(gid) = rx.recv().await {
        debug!("receive gid: {}", gid);

        let task = match aria_rpc.tell_status(gid.clone()).await {
            Ok(task) => task,
            Err(e) => {
                error!("tell_status error: {}", e);
                continue;
            }
        };

        let mut t = match TaskRepo::get_by_gid(&db, gid.clone()).await {
            Ok(Some(t)) => t,
            Ok(None) => continue,
            Err(e) => {
                error!("task get error: {}", e);
                continue;
            }
        };

        t.detail_data = task.detail_data;
        t.status = task.status;
        if t.status == Status::Active && t.detail_data.seeder.is_some_and(|seeder| seeder) {
            t.status = Status::Seeding
        }

        if let Err(e) = TaskRepo::update(&db, t.clone()).await {
            error!("task update error: {}", e)
        }

        let _ = notifier.send(t.clone());

        confirm_follow_task(t, &db, &aria_rpc, notifier.clone()).await;
    }
}

async fn confirm_follow_task(t: Task, db: &DbConn, aria_rpc: &AriaRpc, notifier: Notifier) {
    if t.status != Status::Complete {
        return;
    }

    if t.category != Category::Magnet {
        return;
    }

    if let Some(list) = t.detail_data.followed_by {
        for gid in list {
            let is = match TaskRepo::is_exist(db, gid.clone()).await {
                Ok(is) => is,
                Err(e) => {
                    error!("task is_exist error: {}", e);
                    continue;
                }
            };
            if is {
                continue;
            }

            let task = match add_new_task(db, aria_rpc, gid).await {
                Ok(t) => t,
                Err(e) => {
                    error!("add_new_task error: {}", e);
                    continue;
                }
            };

            let _ = notifier.send(task);
        }
    }
}

async fn add_new_task(db: &DbConn, aria_rpc: &AriaRpc, gid: String) -> Result<Task> {
    let mut t = aria_rpc.tell_status(gid.clone()).await?;

    t.category = Category::Magnet;
    t.created_at = Local::now().timestamp_millis();

    let id = TaskRepo::create(db, t.clone()).await?;
    info!("create task: {}", id);

    t.id = Some(id);

    Ok(t)
}

async fn query_waiting_tasks(state: &AppState, limit: u64) {
    let tasks = match TaskRepo::query_waiting_tasks(&state.db, limit).await {
        Ok(tasks) => tasks,
        Err(e) => {
            error!("query_waiting_tasks error: {}", e);
            return;
        }
    };

    for mut task in tasks {
        let gid = match task.category {
            Category::File => match state
                .aria_rpc
                .add_uri(AriaParams {
                    uri: task.uri.clone(),
                    name: Some(task.name.clone()),
                    dir: Some(task.detail_data.dir.clone()),
                    ..Default::default()
                })
                .await
            {
                Ok(gid) => gid,
                Err(e) => {
                    error!("add_uri error: {}", e);
                    continue;
                }
            },
            Category::Magnet => match state
                .aria_rpc
                .add_magnet(AriaParams {
                    uri: task.uri.clone(),
                    dir: Some(task.detail_data.dir.clone()),
                    ..Default::default()
                })
                .await
            {
                Ok(gid) => gid,
                Err(e) => {
                    error!("add_magnet error: {}", e);
                    continue;
                }
            },
            Category::BitTorrent => match state
                .aria_rpc
                .add_torrent(AriaParams {
                    uri: task.uri.clone(),
                    dir: Some(task.detail_data.dir.clone()),
                    ..Default::default()
                })
                .await
            {
                Ok(gid) => gid,
                Err(e) => {
                    error!("add_torrent error: {}", e);
                    continue;
                }
            },
        };

        task.gid = gid.clone();
        task.status = Status::Active;

        if let Err(e) = TaskRepo::update_gid(&state.db, task).await {
            error!("update_gid error: {}", e);
            return;
        }

        if let Err(e) = state.submitter.send(gid).await {
            error!("submitter.send error: {}", e);
        }
    }
}
